﻿namespace DataIntegration.Messaging.Publisher.ActiveMQ
{
    using System.Reflection;
    using Apache.NMS;
    using Apache.NMS.Util;
    using log4net;
    using Utils.Lifecycle.Base;

    public class ActiveMQPublisher : BaseLifecycle, IPublisher
    {
        private static readonly ILog Logger = LogManager.GetLogger(MethodBase.GetCurrentMethod().ReflectedType);

        private readonly IConnectionFactory connectionFactory;

        private IConnection connection;

        private ISession session;

        private IMessageProducer producer;

        public ActiveMQPublisher(IConnectionFactory connectionFactory)
        {
            Logger.Debug("Creating ActiveMQPublisher");
            this.connectionFactory = connectionFactory;
        }

        protected override void Setup()
        {
            Logger.DebugFormat("Setting up ActiveMQPublisher");
            this.connection = this.connectionFactory.CreateConnection();
            this.session = this.connection.CreateSession();
            this.connection.Start();
        }

        public void Publish(string topic, string messageId, string messageText)
        {
            IMessage message = this.session.CreateTextMessage(messageText);
            message.NMSCorrelationID = messageId;
            Logger.InfoFormat("Publishing message=[{0}] to topic=[{1}]", message.NMSCorrelationID, topic);
            IDestination destination = SessionUtil.GetDestination(
                this.session,
                string.Format("topic://{0}", topic),
                DestinationType.Topic);
            this.producer = this.session.CreateProducer(destination);
            this.producer.DeliveryMode = MsgDeliveryMode.Persistent;
            Logger.DebugFormat("Publishing message [{0}]", messageText);
            this.producer.Send(message);
        }

        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        /// <filterpriority>2</filterpriority>
        public override void Dispose()
        {
            this.Dispose(this.producer);
            this.Dispose(this.session);
            this.Dispose(this.connection);
        }
    }
}